package com.kamistoat.meimeimall.meimeimallpayment.service.impl;

import com.kamistoat.meimeimall.common.constant.DatabaseConstant;
import com.kamistoat.meimeimall.common.constant.PaymentConstant;
import com.kamistoat.meimeimall.common.exception.BizCodeEnum;
import com.kamistoat.meimeimall.common.exception.RRException;
import com.kamistoat.meimeimall.meimeimallpayment.dao.PaymentBillJoinDao;
import com.kamistoat.meimeimall.meimeimallpayment.dao.ReconciliationErrorDao;
import com.kamistoat.meimeimall.meimeimallpayment.entity.PaymentBillJoinEntity;
import com.kamistoat.meimeimall.meimeimallpayment.entity.ReconciliationErrorEntity;
import com.kamistoat.meimeimall.meimeimallpayment.entity.ReconciliationTaskEntity;
import com.kamistoat.meimeimall.meimeimallpayment.service.PaymentBillJoinService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;

@Service("paymentBillJoinService")
public class PaymentBillJoinServiceImpl implements PaymentBillJoinService {
    protected Logger logger = LoggerFactory.getLogger(PaymentBillJoinServiceImpl.class);

    @Qualifier("executeJoinReconciliationThreadPoolExecutor")
    @Autowired
    ThreadPoolExecutor executeJoinReconciliationThreadPoolExecutor;

    @Autowired
    PaymentBillJoinDao paymentBillJoinDao;

    @Autowired
    ReconciliationErrorDao reconciliationErrorDao;

    /**
     * 制作本地交易记录+第三方账单联合大表
     *
     * @param reconciliationTask 任务
     */
    @Override
    @Transactional
    public String joinPaymentAndBillTable(ReconciliationTaskEntity reconciliationTask) {
        String billTableName = PaymentConstant.BILL_RECORD_PREFIX + reconciliationTask.getId();
        String joinTableName = PaymentConstant.PAYMENT_BILL_JOIN_PREFIX + reconciliationTask.getId();

        /**
         * 建关联大表
         */
        try {
            logger.info("创建本地交易记录-第三方账单联合表{}", joinTableName);
            paymentBillJoinDao.createTableWithName(joinTableName);
        } catch (Exception exception) {
            throw new RRException(BizCodeEnum.CREATE_PAYMENT_BILL_JOIN_TABLE_ERROR.getMessage() + ":" + exception, BizCodeEnum.CREATE_PAYMENT_BILL_JOIN_TABLE_ERROR.getCode());
        }

        /**
         * 关联数据
         */
        try {
            logger.info("关联本地交易记录-第三方账单至联合表{}", joinTableName);
            paymentBillJoinDao.joinPaymentAndBill(joinTableName, billTableName, reconciliationTask.getStartTime(), reconciliationTask.getEndTime());
        } catch (Exception exception) {
            throw new RRException(BizCodeEnum.JOIN_PAYMENT_BILL_DATA_ERROR.getMessage() + ":" + exception, BizCodeEnum.JOIN_PAYMENT_BILL_DATA_ERROR.getCode());
        }

        logger.info("数据关联成功，目标数据表{}", joinTableName);
        return joinTableName;

    }

    /**
     * 开始在关联大表上执行对账
     *
     * @param reconciliationTask 任务
     * @return 错误数
     */
    @Override
    @Transactional
    public Long executeReconciliationOnJoinTable(ReconciliationTaskEntity reconciliationTask) {
        String joinTableName = PaymentConstant.PAYMENT_BILL_JOIN_PREFIX + reconciliationTask.getId();
        String errorTableName = PaymentConstant.RECONCILIATION_ERROR_PREFIX + reconciliationTask.getId();
        logger.info("对关联表{}执行对账检查", joinTableName);

        try {
            logger.info("创建错误记录数据表{}", errorTableName);
            reconciliationErrorDao.createTableWithName(errorTableName);
        } catch (Exception exception) {
            throw new RRException(BizCodeEnum.ERROR_TABLE_CREATE_FAIL.getMessage() + ":" + exception, BizCodeEnum.ERROR_TABLE_CREATE_FAIL.getCode());
        }

        logger.info("开启线程对关联表{}执行对账检查", joinTableName);
        int pageSize = PaymentConstant.JOIN_TABLE_PAGE_SIZE;
        long pageOffset = 0L;
        // 分页查询
        List<PaymentBillJoinEntity> joinList = paymentBillJoinDao.selectPage(joinTableName, pageSize, pageOffset);
        /**
         * 从数据库while分页获取数据
         */
        while (joinList != null && joinList.size() > 0) {
            pageOffset += pageSize;
            int sizeOffset = 0;
            List<CompletableFuture<Void>> asyncList = new ArrayList<>();
            /**
             * while拆分join大表，交给各个线程去做
             */
            while (sizeOffset < joinList.size()) {
                List<PaymentBillJoinEntity> subList =
                        new ArrayList<>(joinList.subList(sizeOffset, Math.min(sizeOffset + PaymentConstant.JOIN_DATA_PERTHREAD_SIZE, joinList.size())));
                asyncList.add(CompletableFuture.runAsync(() -> {
                    for (PaymentBillJoinEntity join : subList) {
                        // 正常
                        if (join.getLocThirdPaymentTradeSn() != null && join.getBillThirdPaymentTradeSn() != null &&
                                join.getLocTotalAmount().compareTo(join.getBillTotalAmount()) == 0) {
                            continue;
                        }

                        // 异常
                        ReconciliationErrorEntity errorEntity = new ReconciliationErrorEntity();
                        BeanUtils.copyProperties(join, errorEntity);
                        errorEntity.setId(null);
                        errorEntity.setReconciliationTaskId(reconciliationTask.getId());
                        errorEntity.setHandleStatus(PaymentConstant.ReconciliationErrorHandleStatus.WAITING.getCode());
                        if (join.getLocThirdPaymentTradeSn() == null && join.getBillThirdPaymentTradeSn() != null) {
                            // 长款
                            errorEntity.setErrorType(PaymentConstant.ReconciliationErrorType.LONGSIDE.getCode());
                        } else if (join.getLocThirdPaymentTradeSn() != null && join.getBillThirdPaymentTradeSn() == null) {
                            // 短款
                            errorEntity.setErrorType(PaymentConstant.ReconciliationErrorType.SHORTSIDE.getCode());
                        } else if (join.getLocThirdPaymentTradeSn() != null && join.getBillThirdPaymentTradeSn() != null &&
                                join.getLocTotalAmount().compareTo(join.getBillTotalAmount()) != 0) {
                            // 金额错误
                            errorEntity.setErrorType(PaymentConstant.ReconciliationErrorType.DISMATCH.getCode());
                        } else {
                            // 未知
                            errorEntity.setErrorType(PaymentConstant.ReconciliationErrorType.UNKNOWN.getCode());
                        }
                        reconciliationErrorDao.insertWithName(errorEntity, errorTableName);
                    }
                    subList.clear();
                }, executeJoinReconciliationThreadPoolExecutor));
                sizeOffset += PaymentConstant.JOIN_DATA_PERTHREAD_SIZE;
            }

            try {
                CompletableFuture.allOf(asyncList.toArray(new CompletableFuture[0])).get();
            } catch (Exception exception) {
                throw new RRException(BizCodeEnum.RECONCILIATION_THREAD_EXECUTE_ERROR.getMessage() + ":" + exception, BizCodeEnum.RECONCILIATION_THREAD_EXECUTE_ERROR.getCode());
            }

            /**
             * 当前分页数据全部核对完。开启下一页拉取
             */
            joinList.clear();
            joinList = paymentBillJoinDao.selectPage(joinTableName, pageSize, pageOffset);
        }

        logger.info("关联表{}对账检查完成", joinTableName);

        return reconciliationErrorDao.count(errorTableName);
    }
}
